package com.jiunuan.live.user.provider.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSON;
import com.jiunuan.live.common.interfaces.topic.UserProviderTopicNames;
import com.jiunuan.live.common.interfaces.utils.ConvertBeanUtils;
import com.jiunuan.live.framework.redis.starter.key.UserProviderCacheKeyBuilder;
import com.jiunuan.live.user.constants.CacheAsyncDeleteCode;
import com.jiunuan.live.user.dto.UserCacheAsyncDeleteDTO;
import com.jiunuan.live.user.dto.UserDTO;
import com.jiunuan.live.user.provider.dao.mapper.IUserMapper;
import com.jiunuan.live.user.provider.dao.po.UserPO;
import com.jiunuan.live.user.provider.service.IUserService;
import jakarta.annotation.Resource;
import org.apache.rocketmq.client.producer.MQProducer;
import org.apache.rocketmq.common.message.Message;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author: jiunuan
 * @date: 2024/7/29
 */
@Service
public class UserServiceImpl implements IUserService {
    @Resource
    private IUserMapper userMapper;
    @Resource
    private RedisTemplate<String, UserDTO> redisTemplate;
    @Resource
    private UserProviderCacheKeyBuilder cacheKeyBuilder;
    @Resource
    private MQProducer mqProducer;
    @Override
    public UserDTO getUserById(Long userId) {
        // 检查用户ID是否为空，如果为空，则直接返回null，表示未找到用户信息
        if(userId == null) {
            return null;
        }

        // 根据用户ID构建用户信息的缓存键
        String key = cacheKeyBuilder.buildUserInfoKey(userId);

        // 尝试从Redis缓存中获取用户信息
        UserDTO userDTO = redisTemplate.opsForValue().get(key);
        // 如果缓存中存在用户信息，则直接返回该信息
        if(userDTO != null) {
            return userDTO;
        }

        // 如果缓存中不存在用户信息，则从数据库中查询用户信息
        userDTO = BeanUtil.copyProperties(userMapper.selectById(userId), UserDTO.class);
        // 如果数据库中查询到用户信息，则将其缓存到Redis中，并设置过期时间为30分钟
        if(userDTO != null) {
            redisTemplate.opsForValue().set(key, userDTO, 30L, TimeUnit.MINUTES);
        }

        // 返回从数据库中查询到的用户信息，如果未查询到，则返回null
        return userDTO;

    }

    /**
     * 更新用户信息
     * 此方法主要用于更新数据库中的用户信息，并在更新成功后删除缓存中的用户信息，以保持数据的一致性
     *
     * @param userDTO 包含待更新用户信息的数据传输对象
     * @return 如果用户信息更新成功，返回true；如果用户DTO为空或用户ID为空，则返回false
     */
    @Override
    public boolean updateUserInfo(UserDTO userDTO) {
        // 检查传入的用户信息是否为空或用户ID是否为空，如果是，则直接返回false
        if (userDTO == null || userDTO.getUserId() == null) {
            return false;
        }

        // 调用映射器方法更新数据库中的用户信息，并获取更新状态
        int updateStatus = userMapper.updateById(ConvertBeanUtils.convert(userDTO, UserPO.class));

        // 如果更新状态成功（即影响的行数大于-1），则进行缓存操作
        if (updateStatus > -1) {
            // 构建缓存键并删除缓存中的用户信息
            String key = cacheKeyBuilder.buildUserInfoKey(userDTO.getUserId());
            redisTemplate.delete(key);

            // 准备异步删除缓存所需的信息
            UserCacheAsyncDeleteDTO userCacheAsyncDeleteDTO = new UserCacheAsyncDeleteDTO();
            userCacheAsyncDeleteDTO.setCode(CacheAsyncDeleteCode.USER_INFO_DELETE.getCode());
            Map<String,Object> jsonParam = new HashMap<>();
            jsonParam.put("userId",userDTO.getUserId());
            userCacheAsyncDeleteDTO.setJson(JSON.toJSONString(jsonParam));

            // 创建消息对象，设置主题和消息体，准备发送消息以异步方式删除缓存
            Message message = new Message();
            message.setTopic(UserProviderTopicNames.CACHE_ASYNC_DELETE_TOPIC);
            message.setBody(JSON.toJSONString(userCacheAsyncDeleteDTO).getBytes());
            // 设置延迟级别，使消息发送后一秒再进行缓存删除操作
            message.setDelayTimeLevel(1);

            // 尝试发送消息，如果发送过程中发生异常，则抛出运行时异常
            try {
                mqProducer.send(message);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }

        // 更新操作完成，返回true
        return true;
    }

    @Override
    public boolean insertOne(UserDTO userDTO) {
        if(userDTO == null || userDTO.getUserId() == null) {
            return false;
        }
        userMapper.insert(BeanUtil.copyProperties(userDTO, UserPO.class));
        return true;
    }

    @Override
    public Map<Long, UserDTO> batchQueryUserInfo(List<Long> userIdList) {
        if(CollectionUtils.isEmpty(userIdList)) {
            return Collections.emptyMap();
        }
        //user的id都大于10000
        userIdList = userIdList.stream().filter(id -> id > 10000).collect(Collectors.toList());
        if(CollectionUtils.isEmpty(userIdList)) {
            return Collections.emptyMap();
        }

        //先查询Redis缓存
        List<String> multiKeyList = userIdList.stream()
                .map(userId -> cacheKeyBuilder.buildUserInfoKey(userId)).collect(Collectors.toList());
        List<UserDTO> userDTOList = redisTemplate.opsForValue().multiGet(multiKeyList).stream().filter(x -> x != null).collect(Collectors.toList());
        //若Redis查询出来的数据数量和要查询的数量相等，直接返回
        if(!CollectionUtils.isEmpty(userDTOList) && userDTOList.size() == userIdList.size()) {
            return userDTOList.stream().collect(Collectors.toMap(UserDTO::getUserId, userDTO -> userDTO));
        }
        //不相等，去MySQL查询无缓存的数据
        List<Long> userIdInCacheList = userDTOList.stream().map(UserDTO::getUserId).collect(Collectors.toList());
        List<Long> userIdNotInCacheList = userIdList.stream().filter(userId -> !userIdInCacheList.contains(userId)).collect(Collectors.toList());
        //为了防止sharding-jdbc笛卡尔积路由，对id进行分组
        Map<Long, List<Long>> userIdMap = userIdNotInCacheList.stream().collect(Collectors.groupingBy(userId -> userId % 100));
        List<UserDTO> dbQueryList = new CopyOnWriteArrayList<>();
        userIdMap.values().parallelStream().forEach(queryUserIdList -> {
            dbQueryList.addAll(BeanUtil.copyToList(userMapper.selectBatchIds(queryUserIdList), UserDTO.class));
        });
        //查询MySQL不为空，缓存进Redis
        if(!CollectionUtils.isEmpty(dbQueryList)) {
            Map<String, UserDTO> multiSaveMap = dbQueryList.stream().collect(Collectors.toMap(userDTO -> cacheKeyBuilder.buildUserInfoKey(userDTO.getUserId()), x -> x));
            redisTemplate.opsForValue().multiSet(multiSaveMap);
            //mset不能设置过期时间，使用管道设置，减少网路IO
            redisTemplate.executePipelined(new SessionCallback<Object>() {
                @Override
                public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
                    for (String key : multiSaveMap.keySet()) {
                        operations.expire((K) key, createRandomExpireTime(), TimeUnit.SECONDS);
                    }
                    return null;
                }
            });
            userDTOList.addAll(dbQueryList);
        }
        return userDTOList.stream().collect(Collectors.toMap(UserDTO::getUserId, userDTO -> userDTO));
    }


    /**
     * 生成随机过期时间，单位：秒
     * @return 随机过期时间
     */
    private long createRandomExpireTime() {
        return ThreadLocalRandom.current().nextLong(1000) + 60 * 30;//30min + 1000s
    }
}